package com.tornainbow;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author mcrysler
 */
public class Consumer {
    // Define a logger for this class
    private static final Logger logger = LoggerFactory.getLogger(Consumer.class);
    
    /**
     * Consumes a message which in this case is a Record object
     * @return The Record object that was received
     */
    public static Record receive(String url, String queueName) {
        // URL of the JMS server
        if (url == null || url.trim().isEmpty()) {
            url = ActiveMQConnection.DEFAULT_BROKER_URL;
        }
        logger.debug("URL has been set to {}", url);
        
        // Name of the queue we will be receiving messages from
        if (queueName == null || queueName.trim().isEmpty()) {
            queueName = "TESTQUEUE";
        }
        logger.debug("Queue name has been set to {}", queueName);
        
        try {
            // Getting JMS connection from the server
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            Connection connection = connectionFactory.createConnection();
            connection.start();
            logger.debug("Successfully made connection to {}", url);

            // Creating session for sending messages
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            logger.debug("Successfully created session from the connection");

            // Getting the queue
            Destination destination = session.createQueue(queueName);
            logger.debug("Successfully created queue {}", queueName);

            // MessageConsumer is used for receiving (consuming) messages
            MessageConsumer consumer = session.createConsumer(destination);
            logger.debug("Successfully created consumer");
            
            // Here we receive the message. By default this call is blocking, which
            // means it will wait for a message to arrive on the queue
            Message message = consumer.receive();
            logger.debug("Received message from the queue");

            // There are many types of Message and ObjectMessage is just one of them.
            // Producer sent us an ObjectMessage so we must cast to it to get access
            // to its .getObject() method
            if (message instanceof ObjectMessage) {
                ObjectMessage objectMessage = (ObjectMessage) message;
                logger.debug("Successfully cast message to ObjectMessage");
                
                // If the object returned from the .getObject() call is the object
                // we were expecting then cast to it
                if (objectMessage.getObject() instanceof  Record) {
                    // Since we can assume we've received the message in its entirety
                    // let's call .commit() on the Session object so the message can
                    // be considered as processed
                    Record record = (Record) objectMessage.getObject();
                    
                    // Close the Connection object
                    connection.close();
                    logger.debug("Successfully closed connection");
                    
                    // Return the Record object
                    return record;
                }
                logger.warn("Received an unexpected ObjectMessage from the queue");
                return null;
            } else {
                logger.warn("Received an unexpected type of Message object");
                return null;
            }
        } catch (JMSException e) {
            logger.error("Failed to receive the message from the queue", e);
            return null;
        }
    }
}
